package cn.azzhu.day03
import cn.azzhu.day02.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
/**
 * @author azzhu
 * @create 2020-09-21 10:24:52
 */
object WindowExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.addSource(new SensorSource)

    val keyedStream = stream.keyBy(_.id)

    val windowedStream = keyedStream.timeWindow(Time.seconds(10),Time.seconds(5))

    val reducedStream = windowedStream.reduce((r1, r2) => SensorReading(r1.id, 0L, r1.temperature.min(r2.temperature)))
    reducedStream.print()

    env.execute("WindowExample")
  }
}
